package com.fanxuankai.boot.mqbroker.producer;

import cn.hutool.core.exceptions.ExceptionUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.autoconfigure.MqBrokerProperties;
import com.fanxuankai.boot.mqbroker.producer.autoconfigure.MqBrokerProducerProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DuplicateKeyException;

import javax.annotation.Resource;
import java.sql.SQLIntegrityConstraintViolationException;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * 事件发布者
 *
 * @param <T> 事件数据类型
 * @author fanxuankai
 */
public class EventPublisher<T> {
    private static final Logger LOGGER = LoggerFactory.getLogger(EventPublisher.class);
    @Resource
    protected MsgSendService msgSendService;
    @Resource
    private ThreadPoolExecutor threadPoolExecutor;
    @Resource
    private MqBrokerProperties brokerProperties;
    @Resource
    private MqBrokerProducerProperties properties;
    @Resource
    private MqProducer<MsgSend> mqProducer;

    /**
     * 发布单个事件
     *
     * @param event /
     */
    public void publish(Event<T> event) {
        persistence(event, false);
    }

    /**
     * 发布多个事件
     *
     * @param events /
     */
    public void publish(List<Event<T>> events) {
        persistence(events, false);
    }

    /**
     * 发布单个事件
     *
     * @param event /
     * @param async 是否异步
     */
    public void publish(Event<T> event, boolean async) {
        persistence(event, async);
    }

    /**
     * 发布多个事件
     *
     * @param events /
     * @param async  是否异步
     */
    public void publish(List<Event<T>> events, boolean async) {
        persistence(events, async);
    }

    protected void persistence(Event<T> event, boolean async) {
        if (event == null) {
            return;
        }
        MsgSend msgSend = createMessageSend(event);
        if (async) {
            threadPoolExecutor.execute(() -> save(msgSend));
        } else {
            save(msgSend);
        }
    }

    protected void persistence(List<Event<T>> events, boolean async) {
        if (CollectionUtils.isEmpty(events)) {
            return;
        }
        List<MsgSend> msgSends = events.stream().map(this::createMessageSend).collect(Collectors.toList());
        if (async) {
            threadPoolExecutor.execute(() -> save(msgSends));
        } else {
            save(msgSends);
        }
    }

    @SuppressWarnings("unchecked")
    private void save(MsgSend msgSend) {
        boolean saved = false;
        try {
            saved = msgSendService.save(msgSend);
        } catch (Throwable throwable) {
            if (ExceptionUtil.isCausedBy(throwable, DuplicateKeyException.class,
                    SQLIntegrityConstraintViolationException.class)) {
                LOGGER.warn("防重生产", throwable);
            } else {
                throw throwable;
            }
        }
        if (saved) {
            if (Objects.equals(msgSend.getStatus(), SendStatus.SENDING.getCode())) {
                mqProducer.produce(msgSend);
            }
        }
    }

    @SuppressWarnings("unchecked")
    private void save(List<MsgSend> msgSends) {
        boolean saved = false;
        try {
            saved = msgSendService.saveBatch(msgSends);
        } catch (Throwable throwable) {
            if (ExceptionUtil.isCausedBy(throwable, DuplicateKeyException.class,
                    SQLIntegrityConstraintViolationException.class)) {
                LOGGER.warn("防重生产", throwable);
            } else {
                throw throwable;
            }
        }
        if (saved) {
            msgSends.stream().filter(msgSend -> Objects.equals(msgSend.getStatus(), SendStatus.SENDING.getCode()))
                    .forEach(msgSend -> mqProducer.produce(msgSend));
        }
    }

    private MsgSend createMessageSend(Event<?> event) {
        MsgSend msg = new MsgSend();
        msg.setMqGroup(brokerProperties.getGroup());
        msg.setTopic(event.getName());
        msg.setCode(event.getKey());
        msg.setDataKey(event.getDataKey());
        msg.setEffectTime(event.getEffectTime());
        Object data = event.getData();
        if (data instanceof CharSequence) {
            msg.setData(data.toString());
        } else {
            msg.setData(JSON.toJSONString(data));
        }
        msg.setRetry(0);
        Date now = new Date();
        if (msg.getEffectTime() == null || !now.before(msg.getEffectTime())) {
            msg.setStatus(SendStatus.SENDING.getCode());
        } else {
            if (properties.getDelayed().isEnabled()) {
                msg.setStatus(SendStatus.DELAYED.getCode());
            } else {
                msg.setStatus(SendStatus.WAIT.getCode());
            }
        }
        msg.setEffectTime(Optional.ofNullable(event.getEffectTime()).orElse(now));
        Optional.ofNullable(event.getRetryCount()).ifPresent(msg::setRetryCount);
        msg.setCreateDate(now);
        msg.setLastModifiedDate(now);
        return msg;
    }
}
